/*
 * Copyright (C) 2020 FoilPlanet
 *
 */

#include <stdexcept>
#include <boost/throw_exception.hpp>
#include <boost/asio.hpp>
#include <functional>

// #define OCULUS_LISTEN_PORT              1718
#define DEFAULT_SOCKET_NAME             "minicap" 
#define MODULE_TAG                      "video_sink"

#include "anbox/codec/video_sink.h"
#include "anbox/network/delegate_connection_creator.h"
#include "anbox/network/published_socket_connector.h"
#include "anbox/network/local_socket_messenger.h"
#include "anbox/network/tcp_socket_connector.h"
#include "anbox/network/tcp_socket_messenger.h"
#include "anbox/logger.h"
#include "anbox/system_configuration.h"

#include "rk_mpi.h"
#include "rk_venc_cmd.h"
#include "mpp_buffer.h"

using namespace anbox::video;

extern MPP_RET mpp_buffer_group_share(MppBufferGroup group, unsigned int, unsigned long);

static MppApi  *m_mpi   = nullptr;
static MppCtx   m_ctx   = nullptr;

void Sink::pre_init(const std::shared_ptr<anbox::Runtime> &runtime)
{
    MPP_RET ret;
    (void)runtime;

    if (MPP_OK != (ret = mpp_create(&m_ctx, &m_mpi))) {
        ERROR("mpp_create failed ret %d", ret);
        return;
    }

    // share default ion group
    if (MPP_OK != (ret = mpp_buffer_group_share(NULL, O_DIRECT, 0L))) {
        ERROR("mpp_buffer_group_share failed ret %d: %s", ret, strerror(errno));
    }
}

using namespace anbox::codec;
namespace asio          = boost::asio;
namespace network       = anbox::network;

std::shared_ptr<network::Connector>                     VideoSink::m_serv;
std::vector<std::shared_ptr<network::SocketMessenger>>  VideoSink::m_clients;

#define BANNER_VERSION  1
#define BANNER_SIZE     24
#define MAX_MPP_TRIES   3

void hton_uint32(unsigned char *data, int value)
{
    *data++ = (value & 0x000000FF) >> 0;
    *data++ = (value & 0x0000FF00) >> 8;
    *data++ = (value & 0x00FF0000) >> 16;
    *data++ = (value & 0xFF000000) >> 24;
}

size_t prepare_banner(uint8_t **pbuf, uint32_t width, uint32_t height)
{
    // Prepare banner for oculus web clients.
    static uint8_t banner[BANNER_SIZE];
    banner[0] = 1; // BANNER_VERSION;
    banner[1] = (uint8_t) BANNER_SIZE;
    hton_uint32(banner + 2, getpid());
    hton_uint32(banner + 6,  width);
    hton_uint32(banner + 10, height);
    hton_uint32(banner + 14, width);
    hton_uint32(banner + 18, height);
    banner[22] = 0; // orientation;
    banner[23] = 2; // quirks;

    *pbuf = &banner[0];
    return sizeof(banner);
}

VideoSink::VideoSink(const std::shared_ptr<anbox::Runtime> &runtime)
{
    if (m_serv.get() == nullptr) {
      #ifdef OCULUS_LISTEN_PORT
        INFO("Initializing VideoSink on tcp:%d", OCULUS_LISTEN_PORT);
        auto on_client = [this](
            std::shared_ptr<asio::basic_stream_socket<asio::ip::tcp>> const &sock)
        {
            auto client = std::make_shared<network::TcpSocketMessenger>(sock);
            INFO("Client online :%d", client->local_port());
            client->set_no_delay();
            m_clients.push_back(client);

            uint8_t *bbuf;
            size_t blen = prepare_banner(&bbuf, this->m_width, this->m_height);
            client->send_raw((const char *)bbuf, blen);
        };

        VideoSink::m_serv = std::make_shared<network::TcpSocketConnector>(
            asio::ip::address_v4::from_string("0.0.0.0"),
            OCULUS_LISTEN_PORT,
            runtime,
            std::make_shared<network::DelegateConnectionCreator<asio::ip::tcp>>(on_client)
        );
      #else // !OCULUS_LISTEN_PORT
        const std::string &sockpath = (boost::format("%1%/%2%") %
            SystemConfiguration::instance().socket_dir() % DEFAULT_SOCKET_NAME).str();

        INFO("Initializing VideoSink on unix:%s", sockpath.c_str());
        auto on_client = [this](
            std::shared_ptr<asio::local::stream_protocol::socket> const &sock)
        { 
            auto client = std::make_shared<network::LocalSocketMessenger>(sock);
            INFO("Client online :%d", client->local_port());
            client->set_no_delay();
            m_clients.push_back(client);

            uint8_t *bbuf;
            size_t blen = prepare_banner(&bbuf, this->m_width, this->m_height);
            client->send_raw((const char *)bbuf, blen);
        };

        VideoSink::m_serv = std::make_shared<network::PublishedSocketConnector>(
            sockpath,
            runtime,
            std::make_shared<network::DelegateConnectionCreator<asio::local::stream_protocol>>(on_client)
        );
      #endif // OCULUS_LISTEN_PORT
    }
}

VideoSink::~VideoSink()
{
    if (m_mpi) {
        stream_reset();
    }

    if (m_ctx) {
        mpp_destroy(m_ctx);
        m_ctx = nullptr;
        m_mpi = nullptr;
    }
}

static size_t 
parse_token(const uint8_t *pch_begin, const uint8_t *pch_end, char *out, size_t outsiz)
{
    const uint8_t *pch;
    char *pout = out; 
    const char *outend = out + outsiz - 1;

    for (pch = pch_begin; pch < pch_end && pout < outend; pout++) {
        std::uint8_t ch = *pch++;
        if (ch == ':' || ch == '\0') {
            break;
        }
        *pout = ch;
    }
    *pout = '\0';
    
    // INFO("parse_token '%s' %d bytes", out, pch - pch_begin);
    return pch - pch_begin;
}

void VideoSink::process_data(const uint8_t *data, size_t size)
{
    const uint8_t *pch  = data;
    const uint8_t *pend = data + size;

    // INFO("process_data %d bytes: %s", size, data);

    if (nullptr == m_ctx) {
        ERROR("mpp context not intialized");
    } else {
        stream_reset();
    }

    while (pch < pend) {
        char t1[8], t2[8];
        data = pch; // for dump
        pch += parse_token(pch, pend, &t1[0], sizeof(t1));
        pch += parse_token(pch, pend, &t2[0], sizeof(t2));
        int prim = atoi(t1);
        int len  = atoi(t2);

        switch ((MppRvpuPrimCode)prim) {
        case RVPU_PRIM_INIT:
            stream_init(const_cast<uint8_t *>(pch));
            break;
 
        case RVPU_PRIM_DEINIT:
            stream_end();
            break;

        case RVPU_PRIM_REGS:
            break;

        case RVPU_PRIM_START:
            encode_packet(const_cast<uint8_t *>(pch), len);
            break;

        case RVPU_PRIM_WAIT:
            wait_packet(const_cast<uint8_t *>(pch), len);
            break;

        case RVPU_PRIM_CONTROL_PREP:
            if (len != sizeof(MppEncPrepCfg)) {
                WARNING("Invalid primitive %d: invalid param size %d", prim, len);
            } else {
                stream_control(RVPU_PRIM_CONTROL_PREP, const_cast<uint8_t *>(pch));
            }
            break;

        case RVPU_PRIM_CONTROL_RC:
            if (len != sizeof(MppEncRcCfg)) {
                WARNING("Invalid primitive %d: invalid param size %d", prim, len);
            } else {
                stream_control(RVPU_PRIM_CONTROL_RC, const_cast<uint8_t *>(pch));
            }
            break;

        case RVPU_PRIM_CONTROL_CODEC:
            if (len != sizeof(MppEncH264Cfg)) {
                WARNING("Invalid primitive %d: invalid param size %d", prim, len);
            } else {
                stream_control(RVPU_PRIM_CONTROL_CODEC, const_cast<uint8_t *>(pch));
            }
            break;

        case RVPU_PRIM_CONTROL_SEI:
            if (len != sizeof(MppEncSeiMode)) {
                WARNING("Invalid primitive %d: invalid param size %d", prim, len);
            } else {
                stream_control(RVPU_PRIM_CONTROL_SEI, const_cast<uint8_t *>(pch));
            }
            break;

        case RVPU_PRIM_CONTROL:
            // TODO
            break;

        case RVPU_PRIM_RESET:
            stream_reset();
            break;

        case RVPU_PRIM_FLUSH:
            stream_flush();
            break;

        default:
            WARNING("Unknown primitive '%d' len %d: %s", prim, len, data);
            break;
        }
        
        pch += len;
    } // while (pch < pend)

    if (pch > pend) {
        WARNING("Recv bytes %d, parsed %d", size, pend - pch);
    }
}

void VideoSink::stream_init(void *cfg)
{
    MPP_RET ret;
    (void)cfg;

    MppCodingType type = MPP_VIDEO_CodingAVC;   // H.264/AVC

    if (MPP_OK != (ret = mpp_init(m_ctx, MPP_CTX_ENC, type))) {
        ERROR("mpp_init failed ret %d\n", ret);
        return;
    }

    INFO("mpp_init succuss.");
}

void VideoSink::stream_control(MppRvpuPrimCode prim, void *cfg)
{
    switch (prim) {
    case RVPU_PRIM_CONTROL_PREP: {
        MppEncPrepCfg *prep = static_cast<MppEncPrepCfg *>(cfg);
        INFO("preferences %dx%d format %d", prep->width, prep->height, prep->format);
        m_width  = prep->width;
        m_height = prep->height;
        m_format = prep->format;
        m_hor_stride = prep->hor_stride;
        m_ver_stride = prep->ver_stride;
        (void)m_mpi->control(m_ctx, MPP_ENC_SET_PREP_CFG, prep);
        break;
    }
    case RVPU_PRIM_CONTROL_RC: {
        MppEncRcCfg *prc = static_cast<MppEncRcCfg *>(cfg);
        INFO("rc_mode %s, fps in/out %d/%d", 
             (prc->rc_mode == MPP_ENC_RC_MODE_VBR) ? "VBR" : "CBR", 
             prc->fps_in_num, prc->fps_out_num);
        (void)m_mpi->control(m_ctx, MPP_ENC_SET_RC_CFG, prc);
        break;
    }
    case RVPU_PRIM_CONTROL_CODEC: {
        MppEncH264Cfg *pcfg = static_cast<MppEncH264Cfg *>(cfg);
        INFO("H264 profile %d level %d, qp(init %d max %d min %d)", 
             pcfg->profile, pcfg->level, 
             pcfg->qp_init, pcfg->qp_max, pcfg->qp_min);
        (void)m_mpi->control(m_ctx, MPP_ENC_SET_CODEC_CFG, pcfg);
        break;
    }
    case RVPU_PRIM_CONTROL_SEI: {
        MppEncSeiMode *sei = static_cast<MppEncSeiMode *>(cfg);
        INFO("Sei mode %d", *sei);
        (void)m_mpi->control(m_ctx, MPP_ENC_SET_SEI_CFG, sei);
        break;
    }
    default:
        assert(0);
        break;
    } // (prim)
}

void VideoSink::stream_reset()
{
    MPP_RET ret;
    if (MPP_OK != (ret = m_mpi->reset(m_ctx))) {
        ERROR("mpp reset: %d", ret);
    }
}

void VideoSink::stream_flush()
{
    // NOTHING
}

void VideoSink::stream_end()
{
    // NOTHING
}

void VideoSink::encode_packet(void *data, size_t size)
{
    MPP_RET   ret;
    MppBuffer frmbuf;
    MppFrame  frame  = nullptr;

    MppBufferInfo *info = static_cast<MppBufferInfo *>(data);

    if (size != sizeof(*info)) {
        ERROR("Mismatched type size %d", size);
        return;
    }

    if (MPP_OK != (ret = mpp_buffer_import(&frmbuf, info))) {
        ERROR("mpp_buffer_import (type:%d size:%06X fd:%d hnd:%p ptr:%p) failed: %d", 
              info->type, info->size, info->fd, info->hnd, info->ptr, ret);
        return;
    }

    if (MPP_OK != (ret = mpp_frame_init(&frame))) {
        ERROR("mpp_frame_init failed");
        mpp_buffer_put(frmbuf);
        return;
    }

#if 1
    mpp_frame_set_width(frame, m_width);
    mpp_frame_set_height(frame, m_height);
    mpp_frame_set_hor_stride(frame, m_hor_stride);
    mpp_frame_set_ver_stride(frame, m_ver_stride);
    mpp_frame_set_fmt(frame, static_cast<MppFrameFormat>(m_format));
#endif
    
    DEBUG("frame %dx%d", m_width, m_height);

    mpp_frame_set_buffer(frame, frmbuf);
    mpp_frame_set_eos(frame, (frmbuf == nullptr) ? 1 : 0);

    if (MPP_OK != (ret = m_mpi->encode_put_frame(m_ctx, frame))) {
        ERROR("mpp encode_put_frame: %d", ret);
    }
    
    // mpp_frame_deinit(&frame);
    mpp_buffer_put(frmbuf);
}

void VideoSink::wait_packet(void *data, size_t size)
{
    MPP_RET   ret;
    MppBufferInfo *info = static_cast<MppBufferInfo *>(data);
    MppPacket packet = nullptr;

    if (size != sizeof(*info)) {
        ERROR("Mismatched type size %d", size);
        return;
    }

    int n = 0;
    do {
        if (MPP_OK != (ret = m_mpi->encode_get_packet(m_ctx, &packet))) {
            ERROR("mpp encode_get_packet: %d", ret);
            return;
        }

        if (nullptr == packet) {
            ERROR("encode_get_packet timeout %d", ++n);
            std::this_thread::sleep_for(std::chrono::milliseconds(1)); // usleep(1000);
        }
    } while (nullptr == packet && n < MAX_MPP_TRIES);

    if (nullptr != packet) {
        // send to peer
        void  *pkt_ptr = mpp_packet_get_pos(packet);
        size_t pkt_len = mpp_packet_get_length(packet);

        DEBUG("packet %d bytes broadcast %d", pkt_len, m_clients.size());

        if (m_clients.size() > 0) {
            std::vector<std::shared_ptr<network::SocketMessenger>> offlines;
            MppBuffer buffer;
            mpp_buffer_get(NULL, &buffer, pkt_len + 4);
            uint8_t *phead = (uint8_t *)mpp_buffer_get_ptr(buffer);
            hton_uint32(phead, pkt_len);
            memcpy(phead + sizeof(uint32_t), pkt_ptr, pkt_len);
            for (auto peer : m_clients) {
                if (peer->send_raw((const char *)phead, pkt_len + 4) <= 0) {
                    WARNING("Client offline :%d", peer->local_port());
                    offlines.push_back(peer);
                }
            }
            mpp_buffer_put(buffer);

            // remove offline clients
            if (offlines.size() > 0) {
                for (auto peer : offlines) {
                    for (auto it = m_clients.begin(); it < m_clients.end(); it++) {
                        if (*it == peer) {
                            peer->close();
                            m_clients.erase(it);
                            DEBUG("Client :%d removed", peer->local_port());
                            break;
                        }
                    }
                }
            }
        }
        // release packet
        mpp_packet_deinit(&packet);
    }
}